package org.acghub.mtdb.core;

import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import org.acghub.mtdb.common.Config;
import org.acghub.mtdb.common.Lifecycle;
import org.acghub.mtdb.common.exception.DeadLockError;
import org.acghub.mtdb.common.exception.LifecycleException;
import org.acghub.mtdb.common.exception.MtdbError;
import org.acghub.mtdb.common.exception.MtdbException;

import java.util.Collection;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.*;

@Slf4j
final class Procedures implements Lifecycle<Config> {

    static Procedures INSTANCE = new Procedures();

    private ProcedureExecutor[] executors;
    private int sizeMask;

    private HashedWheelTimer timer;

    private ScheduledExecutorService scheduler;

    static ProcedureExecutor wrap(ExecutorService executorService) {
        return new DefaultProcedureExecutor(executorService);
    }

    public ProcedureExecutor selectProcedureExecutor() {
        int i = ThreadLocalRandom.current().nextInt(0, sizeMask + 1);
        return executors[i];
    }

    public ProcedureExecutor selectProcedureExecutor(Object key) {
        if (Objects.isNull(key)) {
            return selectProcedureExecutor();
        }
        return executors[key.hashCode() % sizeMask];
    }

    @Override
    public void init(Config config) throws LifecycleException {
        int procedureExecutorShift = config.getProcedureExecutorShift();
        if (procedureExecutorShift < 0 || procedureExecutorShift > 8) {
            throw new LifecycleException("procedure executor size must be greater equal than 0 and less equal than 8, procedureExecutorShift: " + procedureExecutorShift);
        }
        int procedureExecutorSize = 1 << procedureExecutorShift;
        sizeMask = procedureExecutorSize - 1;
        executors = new ProcedureExecutor[procedureExecutorShift];
        for (int i = 0; i < procedureExecutorShift; i++) {
            executors[i] = DefaultProcedureExecutor.wrap(Executors.newSingleThreadExecutor());
        }
        timer = new HashedWheelTimer();
        scheduler = Executors.newScheduledThreadPool(4);
    }

    @Override
    public void close() throws LifecycleException {
        try {
            Set<Timeout> cancelled = timer.stop();
            log.info("timer stop, canceled.size: {}", cancelled.size());
        } catch (Exception e) {
            log.error("timer stop error", e);
        }

        boolean terminated = false;
        try {
            scheduler.shutdown();
            terminated = scheduler.awaitTermination(1, TimeUnit.MINUTES);
        } catch (Exception e) {
            log.error("scheduler awaitTermination error", e);
        } finally {
            if (!terminated) {
                try {
                    scheduler.shutdownNow();
                } catch (Exception ex) {
                    log.error("scheduler shutdownNow error", ex);
                }
            }
        }
    }

    private Timeout timeout(TimerTask task, int time, TimeUnit unit) {
        return timer.newTimeout(task, time, unit);
    }

    private Timeout timeout(Runnable runnable, int time, TimeUnit unit) {
        return timeout(t -> runnable.run(), time, unit);
    }

    private Future<?> schedule(Runnable runnable, int time, TimeUnit unit) {
        return scheduler.schedule(runnable, time, unit);
    }

    /**
     * @author zhouhe
     */
    private static final class DefaultProcedureExecutor implements ProcedureExecutor {

        private final ExecutorService executorService;

        public DefaultProcedureExecutor(ExecutorService executorService) {
            this.executorService = executorService;
        }

        private static ProcedureExecutor wrap(ExecutorService executorService) {
            return new DefaultProcedureExecutor(executorService);
        }

        @Override
        public <P extends AbstractProcedure> void execute(P procedure) {
            new DefaultProcedureFuture<P>(procedure, this);
        }

        @Override
        public <P extends AbstractProcedure> ProcedureFuture<P> submit(P procedure) {
            if (Transaction.isActive()) {
                throw new MtdbException("can not submit in transaction.");
            }
            return new DefaultProcedureFuture<P>(procedure, this);
        }

        @Override
        public ExecutorService wrapped() {
            return executorService;
        }
    }

    /**
     * @author zhouhe
     */
    @Slf4j
    private static final class DefaultProcedureFuture<P extends AbstractProcedure> implements ProcedureFuture<P> {

        private final P p;
        private final ProcedureExecutor executor;
        private volatile Future<?> future;
        private volatile int retryTimes = 0;

        public DefaultProcedureFuture(P p, ProcedureExecutor executor) {
            this.p = p;
            this.executor = executor;
            Objects.requireNonNull(p);
            Objects.requireNonNull(executor);
            this.submit();
        }

        @Override
        public void run() {
            try {
                try {
                    Transaction.create().perform(p);
                } finally {
                    Transaction.destroy();
                }
                this.finished();
            } catch (DeadLockError e) {
                if (this.incrRetryTimes() > p.times()) {
                    this.finished();
                    throw e;
                }
                throw new FutureDeadLock(INSTANCE.schedule(this::submit, p.delay(), TimeUnit.MILLISECONDS));
            } catch (MtdbError e) {
                this.finished();
                throw e;
            } catch (Throwable e) {
                this.finished();
                throw new MtdbError(e);
            }
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return future.cancel(mayInterruptIfRunning);
        }

        @Override
        public Procedure.Result get() throws InterruptedException, ExecutionException {
            // 用异常的方式 等待重试次数完成
            while (true) {
                try {
                    future.get();
                    return p.getResult();
                } catch (ExecutionException e) {
                    if (!(e.getCause() instanceof FutureDeadLock)) {
                        throw new ExecutionException(p.getCause());
                    }
                    ((FutureDeadLock) e.getCause()).pending();
                }
            }
        }

        @Override
        public Procedure.Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            // 用异常的方式 等待重试次数完成
            while (true) {
                try {
                    future.get(timeout, unit);
                    return p.getResult();
                } catch (ExecutionException e) {
                    if (!(e.getCause() instanceof FutureDeadLock)) {
                        throw new ExecutionException(p.getCause());
                    }
                    ((FutureDeadLock) e.getCause()).pending();
                }
            }
        }

        @Override
        public boolean isCancelled() {
            return future.isCancelled();
        }

        @Override
        public boolean isDone() {
            return future.isDone();
        }

        @Override
        public P procedure() {
            return p;
        }

        private int incrRetryTimes() {
            return ++retryTimes;
        }

        private void submit() {
            this.future = executor.wrapped().submit(this);
        }

        private void finished() {
            try {
                Collection<Procedure.ProcedureListener<P>> listeners = p.getListeners();
                if (listeners != null) {
                    for (Procedure.ProcedureListener<P> listener : listeners) {
                        listener.onResult(p, p.getResult());
                    }
                }
            } catch (Throwable e) {
                log.error("notifyListeners error", e);
            }
        }
    }

    private static class FutureDeadLock extends MtdbError {
        private final Future<?> future;

        FutureDeadLock(Future<?> future) {
            this.future = future;
        }

        private void pending() {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException ignored) {
            }
        }
    }
}
